package model

import (
	"fmt"
	"gitee.com/Ljolan/si-mqtt/corev5/authv5/authplus"
	"gitee.com/Ljolan/si-mqtt/corev5/messagev5"
	"gitee.com/Ljolan/si-mqtt/corev5/servicev5"
	"gitee.com/Ljolan/si-mqtt/corev5/topicsv5"
	logger2 "gitee.com/Ljolan/si-mqtt/logger"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

type Type int

func (t Type) String() string {
	if t == Sub {
		return "Sub"
	} else if t == Pub {
		return "Pub"
	} else {
		return "-"
	}
}

const (
	_ Type = iota
	Sub
	Pub
)

type MqttModel struct {
	c       *servicev5.Client
	au      authplus.AuthPlus
	cid     string // 客户端id
	addr    string
	PubNum  int   // pub数量
	PubTime int64 // pub间隔
	pkid    uint32

	pubTopic string
	pubQos   uint64
	payload  string
	subTopic string
	subQos   uint64

	allTopic []string
	Type
}

var once sync.Once

// var mp = make(map[string]bool)
func NewClient(clientId string, addr string, pubNum int, pubTime int64, p Type, pubTopic string,
	pubQos uint64, payload string, subTopic string, subQos uint64) *MqttModel {
	m := &MqttModel{}
	once.Do(func() {
		logger2.LogInit("info") // 日志必须提前初始化
	})
	authplus.Register(clientId, authplus.NewDefaultAuth())
	vp, _ := authplus.NewManager(clientId)
	if clientId == "" {
		clientId = "asdasdasfgfqwe312erfasr12rasd.,//"
	}

	m.c = &servicev5.Client{}
	m.c.AuthPlus = vp
	m.cid = clientId
	m.addr = addr
	m.PubNum = pubNum
	m.PubTime = pubTime
	m.Type = p
	m.pubTopic = pubTopic
	m.pubQos = pubQos
	m.payload = payload
	m.subTopic = subTopic
	m.subQos = subQos
	return m
}
func (m *MqttModel) New(i string, types Type) *MqttModel {
	return NewClient(m.cid+"-"+types.String()+"-"+i, m.addr, m.PubNum, m.PubTime, types,
		m.pubTopic, m.pubQos, m.payload, m.subTopic, m.subQos)
}
func (m *MqttModel) SetAddr(i string) {
	m.addr = i
}

var pkid uint32

func (m *MqttModel) Publish(ch chan struct{}) error {
	// Creates a new PUBLISH messagev5 with the appropriate contents for publishing
	pubmsg := messagev5.NewPublishMessage()
	err := pubmsg.SetTopic([]byte(m.pubTopic))
	if err != nil {
		return err
	}
	pubmsg.SetPayload([]byte(m.payload))
	if m.pubQos > 0 {
		pubmsg.SetPacketId(uint16(atomic.AddUint32(&m.pkid, 1)))
	}
	err = pubmsg.SetQoS(byte(m.pubQos))
	if err != nil {
		return err
	}
	// Publishes to the server by sending the messagev5
	return m.c.Publish(pubmsg, func(msg, ack messagev5.Message, err error) error {
		if err != nil {
			fmt.Println(err)
		} else {
		}
		ch <- struct{}{}
		return nil
	})
}

var incr int64
var size int64

func GetAllRecMsg() int64 {
	return atomic.LoadInt64(&size)
}
func (m *MqttModel) Topic() string {
	return m.subTopic
}
func (m *MqttModel) SubTopic(ch chan struct{}, tps ...string) error {
	submsg := messagev5.NewSubscribeMessage()
	topic := m.subTopic + "_" + strconv.Itoa(int(atomic.AddInt64(&incr, 1)))
	if len(tps) > 0 {
		topic = tps[0]
	}
	err := submsg.AddTopic([]byte(topic), byte(m.subQos))
	if err != nil {
		return err
	}
	return m.c.Subscribe(submsg, func(msg, ack messagev5.Message, err error) error {
		if err != nil {
			fmt.Println(err)
		} else {
			// fmt.Println("\n====== >>> Subscribe msg handle")

		}
		ch <- struct{}{}
		return nil
	}, func(msg *messagev5.PublishMessage, sub topicsv5.Sub, string2 string, bool2 bool) error {
		// fmt.Println("===<<<>>>", msg.String())
		atomic.AddInt64(&size, 1)
		return nil
	})
}

var mutx = sync.Mutex{}

func (m *MqttModel) Disconnect() error {
	time.Sleep(time.Second * 10)
	mutx.Lock()
	defer func() {
		if e := recover(); e != nil {
			fmt.Println(e)
		}
		mutx.Unlock()
	}()
	m.c.Disconnect()
	return nil
}
func (m *MqttModel) UnSubTopic() error {
	for i := 0; i < len(m.allTopic); i++ {
		unsubmsg := messagev5.NewUnsubscribeMessage()
		unsubmsg.AddTopic([]byte(m.allTopic[i]))
		err := m.c.Unsubscribe(unsubmsg, func(msg, ack messagev5.Message, err error) error {
			if err != nil {
				fmt.Println(err)
			} else {
				//fmt.Println("\n====== >>> Unsubscribe handle")
			}
			return nil
		})
		if err != nil {
			return err
		}
	}
	return nil
}
func (m *MqttModel) Connect() error {
	// Creates a new MQTT CONNECT messagev5 and sets the proper parameters
	msg := messagev5.NewConnectMessage()
	msg.SetWillQos(1)
	msg.SetVersion(5)
	msg.SetCleanSession(false)
	err := msg.SetClientId([]byte(m.cid))
	if err != nil {
		return err
	}
	msg.SetKeepAlive(30)
	msg.SetSessionExpiryInterval(25)
	msg.SetWillTopic([]byte("will"))
	msg.SetWillMessage([]byte("send me home"))
	msg.SetUsername([]byte("surgemq"))
	msg.SetPassword([]byte("verysecret"))
	//msg.SetAuthMethod([]byte("default"))
	//msg.SetAuthData([]byte("aaa"))
	go func() {
		for {
			time.Sleep(time.Duration(int(msg.SessionExpiryInterval()*2/3)) * time.Second)
			m.c.Ping(nil)
		}
	}()
	return m.c.Connect(m.addr, msg)
}
